package com.idanchuang.support.etcache.pubsub.pulsar;

import com.idanchuang.support.etcache.pubsub.AbstractRemoteEventPublisher;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于 Pulsar 实现的缓存变更事件发布
 *
 * @author yjy
 * Created at 2021/12/22 11:12 上午
 */
public class PulsarRemoteEventPublisher extends AbstractRemoteEventPublisher {

    private static final Logger logger = LoggerFactory.getLogger(PulsarRemoteEventPublisher.class);

    private final PulsarProperties properties;
    private PulsarClient client;
    private final String topic;
    private Producer<byte[]> producer;

    public PulsarRemoteEventPublisher(String env, String appId, PulsarProperties properties) {
       try {
           this.properties = properties;
           this.topic = env + "-base/etcache/event-" + appId.toLowerCase();
           this.client = PulsarClient.builder().serviceUrl(properties.getServiceUrl()).build();
           this.producer = client.newProducer()
                   .topic(this.topic)
                   .create();
       } catch (Exception e) {
           throw new RuntimeException("PulsarRemoteEventPublisher init failed", e);
       }
    }

    @Override
    public void doPublish(byte[] remoteKeyBytes) throws Exception {
        if (!this.producer.isConnected()) {
            reconnect();
        }
        // 然后你就可以发送消息到指定的broker 和topic上：
        this.producer.send(remoteKeyBytes);
    }

    /**
     * 重连amqp
     */
    private void reconnect() {
        while (!isShutdown() && !this.producer.isConnected()) {
            try {
                Thread.sleep(1000L);
                if (this.client.isClosed()) {
                    this.client = PulsarClient.builder().serviceUrl(properties.getServiceUrl()).build();
                }
                this.producer = client.newProducer().topic(this.topic).create();
                logger.info("reconnect pulsar success, topic: {}", this.topic);
            } catch (InterruptedException e) {
                logger.warn("reconnect pulsar cancel, cause thread is interrupted", e);
                break;
            } catch (Exception e) {
                logger.error("reconnect pulsar failed, will retry after 1s..", e);
            }
        }
    }

    @Override
    public void destroy() {
        try {
            super.destroy();
            if (this.client != null && !this.client.isClosed()) {
                this.client.close();
            }
            logger.info("PulsarRemoteEventSubscriber destroy finish > topic: {}", topic);
        } catch (Exception e) {
            logger.warn("connection close failed, cause: {}", e.getMessage());
        }
    }

    public PulsarProperties getProperties() {
        return properties;
    }

    public PulsarClient getClient() {
        return client;
    }

    public String getTopic() {
        return topic;
    }

    public Producer<byte[]> getProducer() {
        return producer;
    }
}
